package com.gitee.ed.example.reactor;

import org.reactivestreams.Publisher;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

import java.util.List;
import java.util.function.UnaryOperator;
import java.util.stream.Stream;

/**
 * @author hongda.li
 */
public final class FluxExecutor<T> {

    private Flux<T> flux;

    private FluxExecutor(Flux<T> flux) {
        this.flux = flux;
    }

    @SafeVarargs
    public static <T> FluxExecutor<T> create(final T... array) {
        return FluxExecutor.create(Flux.fromArray(array));
    }

    public static <T> FluxExecutor<T> create(final Stream<T> stream) {
        return FluxExecutor.create(Flux.fromStream(stream));
    }

    public static <T> FluxExecutor<T> create(final Iterable<T> collection) {
        return FluxExecutor.create(Flux.fromIterable(collection));
    }

    public static <T> FluxExecutor<T> create(final Publisher<T> publisher) {
        return new FluxExecutor<>(Flux.from(publisher));
    }

    /**
     * 以编程方式创建具有多次发射能力的 Flux
     *
     * @param emitter Flux 发射器
     * @param <T>     Flux 元素泛型
     * @return 创建一个 Flux 执行器
     */
    public static <T> FluxExecutor<T> create(FluxEmitter<T> emitter) {
        return new FluxExecutor<>(Flux.create(emitter::emit));
    }

    /**
     * 以编程方式创建一个的 Flux
     * generator 中 next 只能调 1 次
     * generator 中 index 可以作为元素索引
     *
     * @param generator Flux 生成器
     * @param <T>       Flux 元素泛型
     * @return Flux 执行器
     */
    public static <T> FluxExecutor<T> create(FluxGenerator<T> generator) {
        return new FluxExecutor<>(Flux.generate(() -> 0L, generator::generate));
    }

    /**
     * 直接创建一个 Flux 流
     *
     * @param flux Flux 流
     * @param <T>  Flux 元素泛型
     * @return Flux 执行器
     */
    public static <T> FluxExecutor<T> create(final Flux<T> flux) {
        return new FluxExecutor<>(flux);
    }

    /**
     * 转换 Flux 流
     *
     * @param transform 转换器
     * @return 链式调用
     */
    public FluxExecutor<T> transform(UnaryOperator<Flux<T>> transform) {
        this.flux = transform.apply(this.flux);
        return this;
    }

    /**
     * 将流式转为集合
     *
     * @return 有序集合
     */
    public List<T> toList() {
        return this.flux.toStream().toList();
    }

    /**
     * 将流式转为流
     *
     * @return 流
     */
    public Stream<T> toStream() {
        return this.flux.toStream();
    }

    /**
     * 获取创建完成的 Flux流
     *
     * @return Flux 流
     */
    public Flux<T> getFlux() {
        return flux;
    }

    /**
     * 订阅并执行流
     *
     * @param subscriber 流订阅器
     */
    public void execute(BaseSubscriber<T> subscriber) {
        this.flux.subscribe(subscriber);
    }
}
